package com.arch.flink.biz;

import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @author pizhihui
 * @date 2022-01-07 19:46
 */
public class TimeWindowWordCount2 {

    public static void main(String[] args) throws Exception {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> socketStream = env.socketTextStream("localhost", 8888);



        SingleOutputStreamOperator<Tuple2<String, Integer>> dataStream = socketStream.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String line, Collector<Tuple2<String, Integer>> collector) throws Exception {
                String[] split = line.split(",");
                for (String s : split) {
                    collector.collect(new Tuple2<>(s, 1));
                }
            }
        });

        SingleOutputStreamOperator<Tuple2<String, Integer>> sum = dataStream
                .keyBy(tuple -> tuple.f0)
                //.timeWindow(Time.seconds(10), Time.seconds(5))
                .window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                .process(new SumProcessFunction());



        sum.print().setParallelism(1);

        env.execute("TimeWordCount");


    }

    // IN OUT KEY Window
    public static class SumProcessFunction
            extends ProcessWindowFunction<Tuple2<String, Integer>,
            Tuple2<String, Integer>,
            String, TimeWindow> {

        FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");

        @Override
        public void process(String s, Context context, java.lang.Iterable<Tuple2<String, Integer>> elements, Collector<Tuple2<String, Integer>> out) throws Exception {
            System.out.println("当前系统时间: " + dateFormat.format(System.currentTimeMillis()));
            System.out.println("窗口处理时间: " + dateFormat.format(context.currentProcessingTime()));

            System.out.println("窗口开始时间: " + dateFormat.format(context.window().getStart()));
            System.out.println("窗口结束时间: " + dateFormat.format(context.window().getEnd()));

            System.out.println("==========================================================================" + Thread.currentThread().getName());



            int count = 0;
            for (Tuple2<String, Integer> element : elements) {
                count++;
            }
            out.collect(Tuple2.of(s, count));

        }
    }
}
